﻿#include "UDPProcess.h"
#include "UDPApi.h"
#include <iostream>
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR,12)


#if 0
UDPProcessThread::UDPProcessThread(CodeProcess &code):m_code(code)
{
	m_initsocket = false;
	mediainfo = new MediaInfo();
	memset(m_host,0,sizeof(m_host));
	memset(m_rhost,0,sizeof(m_rhost));
	
}
#endif
UDPProcessThread::UDPProcessThread(MediaInfo *media, UDPApi *udpapi)
{
	m_num = 1;
	m_timestamp = 0;
	m_initsocket = false;
	mediainfo =media;
	mUDPApi = udpapi;
	memset(m_host,0,sizeof(m_host));
	memset(m_rhost,0,sizeof(m_rhost));
	m_port = 0;
	m_rport = 0;
	m_isOpenRing = false;
	m_send_mutex.Init();
	m_recv_mutex.Init();
	m_isSend = false;
	m_isRecv = false;
	m_isSendHeatbeat = false;
	m_isSendPttInfo = false;
	m_isSendEndPttInfo = false;
	m_Open_mutex.Init();
	sendthread = NULL;
	recivethread = NULL;
	m_cur_uid = 0;

	isSendFlag = false;
}

int UDPProcessThread::InitLocal(char* host, unsigned short port)
{
	if (host==NULL||port == 0)
	{
		return -1;
	}
	strcpy(m_host,host);
	m_port = port;
	return 0;
}
int UDPProcessThread::InitRemote(char* rhost, unsigned short rport)
{
	if (rhost==NULL||rport == 0)
	{
		return -1;
	}
	strcpy(m_rhost,rhost);
	m_rport = rport;
	m_destaddr.sin_addr.S_un.S_addr = inet_addr(m_rhost);
	m_destaddr.sin_family = AF_INET;
	m_destaddr.sin_port = htons(m_rport);
	return 0;
}

int UDPProcessThread::CreateThread()
{
	int ret = 0;
	sendthread = new UDPSendThread(*this);
	if (sendthread == 0)
	{
			
		return -1;
	}
	if ((ret = sendthread->Start()) < 0)
	{
		delete sendthread;
		sendthread = NULL;
		return ret ;
	}

	recivethread = new UDPReciveThread(*this);
	if(recivethread == 0)
	{
		
		if (sendthread!=NULL)
		{
			sendthread->Stop();
			delete sendthread;
			sendthread = NULL;
		}
		
		return -2;
	}
	if((ret = recivethread->Start())<0)
	{
		delete recivethread;
		recivethread = NULL;
		if (sendthread!=NULL)
		{
			sendthread->Stop();
			delete sendthread;
			sendthread = NULL;			
		}		
		return ret;	
	}	
	
	return 0;
}
int UDPProcessThread::StopThread()
{
	if (sendthread!=NULL)
	{
		sendthread->Stop();
		delete sendthread;
		sendthread = NULL;
	}

	if (recivethread != NULL)
	{
		recivethread->Stop();
		delete recivethread;
		recivethread = NULL;
	}

	return 0;
}

int UDPProcessThread::UDPServer()
{
	isockaddr = sizeof(SOCKADDR);
	m_socket = socket(AF_INET,SOCK_DGRAM,0);
	printf("m_host:%s\n",m_host);
	m_localaddr.sin_addr.S_un.S_addr = inet_addr(m_host);
	m_localaddr.sin_family = AF_INET;
	m_localaddr.sin_port = htons(m_port);
	int ret = bind(m_socket,(struct sockaddr *)&m_localaddr,sizeof(m_localaddr));
	if (ret == -1)
	{
		printf("bind failed!\n");
		return -1;
	}
	printf("bind is suc!\n");
	return 0;
}
#if 0
int UDPProcessThread::UDPClient()
{
	m_initsocket = true;
	isockaddr = sizeof(SOCKADDR);
	m_socket = socket(AF_INET,SOCK_DGRAM,0);
	m_destaddr.sin_addr.S_un.S_addr = inet_addr(m_ip);
	m_destaddr.sin_family = AF_INET;
	m_destaddr.sin_port = htons(m_cliport);
	return 0;
}
#endif

UDPProcessThread::~UDPProcessThread()
{
	if (m_initsocket && m_socket!= 0)
		closesocket(m_socket);
	m_socket = 0;
	m_initsocket = false;
	//if (mediainfo!=NULL)
	//{
	//	delete mediainfo;
	//	mediainfo = NULL;
	//}
}


int UDPProcessThread::Init()
{
	WORD versionReq;
	WSADATA wsaData;
	int err;
	versionReq = MAKEWORD(2,4);
	err = WSAStartup(versionReq,&wsaData);
	if (err != 0)
	{
		return -1;
	}
	m_initsocket = true;
	m_socket = socket(AF_INET,SOCK_DGRAM,0);
	printf("socket = %d\n",m_socket);
// 	DWORD dwBytesReturned = 0;
// 	BOOL  bNewBehavior = FALSE;
// 	DWORD status;
// 	status = WSAIoctl(m_socket, SIO_UDP_CONNRESET,
// 		&bNewBehavior, sizeof(bNewBehavior),
// 		NULL, 0, &dwBytesReturned,
// 		NULL, NULL);
// 	
// 	if (SOCKET_ERROR == status)
// 	{
// 		DWORD dwErr = WSAGetLastError();
// 		if (WSAEWOULDBLOCK == dwErr)
// 		{
// 			// nothing to do
// 			return -1;
// 		}
// 		else
// 		{
// 			printf("WSAIoctl(SIO_UDP_CONNRESET) Error: %d/n", dwErr);
// 			return -2;
// 		}
// 		
// 	}
	printf("m_host:%s\n",m_host);
	m_localaddr.sin_addr.S_un.S_addr = inet_addr(m_host);
	m_localaddr.sin_family = AF_INET;
	m_localaddr.sin_port = htons(m_port);
	int ret = bind(m_socket,(struct sockaddr *)&m_localaddr,sizeof(m_localaddr));
	if (ret == -1)
	{
		printf("bind failed!\n");
		return -3;
	}
	printf("bind success ret = %d\n",ret);
	m_destaddr.sin_addr.S_un.S_addr = inet_addr(m_rhost);
	m_destaddr.sin_family = AF_INET;
	m_destaddr.sin_port = htons(m_rport);
	/*
	err = UDPServer();
	if (err<0)
	{		
		printf("bind err!\n");
		return err;
	}
	*/
//	printf("sock init!\n");
	return 0;
}

int UDPProcessThread::OpenRing()
{
	
	FILE *fp=NULL;
	char infodata[1024];
	int ret = 0;
	RTP_INFO info;
	int num = 1;
//	char *p;
	int i = 0;
//	RTPHeader *head = NULL;

		memset(&info,0,sizeof(RTP_INFO));
		memset(infodata,0,sizeof(infodata));
		fp = fopen("../2620.wav","rb+");
		if (fp==NULL)
		{
			printf("open file error");
			return -1;
		}
		waveHead aHeader; 
		WAVE_FORMAT m_wformat;
		int m_iDataStart,m_iDataSize;
		memset(&m_wformat,0,sizeof(WAVE_FORMAT));
		fseek(fp, 0, SEEK_SET); 
		fread(&aHeader, 1, sizeof(waveHead), fp); 
		memcpy(&m_wformat, &(aHeader.wavFormat), sizeof(WAVE_FORMAT)); 
		//m_wformat.cbSize = 0; 
		fseek(fp, aHeader.dwFmtSize-16, SEEK_CUR); 
		FACT_BLOCK fact; 
		DATA_BLOCK data; 
		fread(&fact, 1, sizeof(FACT_BLOCK), fp); 
		if (*((DWORD*)fact.szFactID) == *((DWORD*)"fact")) 
		{ 
			fseek(fp, fact.dwFactSize, SEEK_CUR); 
			fread(&data, 1, sizeof(DATA_BLOCK), fp); 
		} 
		else if (*((DWORD*)fact.szFactID) == *((DWORD*)"data")) 
		memcpy(&data, &fact, sizeof(DATA_BLOCK)); 
		m_iDataStart = ftell(fp); 
		m_iDataSize = data.dwDataSize;
		printf("m_idata size = %d\n",m_iDataSize);
// 		p = (char*)malloc(m_iDataSize);

// 		fseek(fp,44L,SEEK_SET);
// 		mediainfo->rtpmutex_put.Lock();
// 		while(!feof(fp))
// 		{
// 			memset(&info,0,sizeof(info));
// 			ret = fread(info.DecData,320,1,fp);
// 			if (ret>0)
// 			{
// 				info.PayLoadLen = 320;
// 				mediainfo->getque.push_back(info);
// 			}
// 		}
// 		mediainfo->rtpmutex_put.Unlock();

		int n = m_iDataSize%320;
		int wavNum;
		if (n==0)
		{
			wavNum=m_iDataSize/320;
		}
		if (n!=0)
		{
			wavNum=(m_iDataSize/320)+1;
		}
		mediainfo->rtpmutex_put.Lock();
		for (int j =1;j<=wavNum;j++)
		{
			fseek(fp,0L,SEEK_CUR);
			memset(infodata,0,sizeof(infodata));
			ret = fread(infodata,1,320,fp);
			if (ret>0)
			{
				memset(&info,0,sizeof(RTP_INFO));
				info.PayLoadLen = 320;
				info.PayLoadType = 0;
				memcpy(info.DecData,infodata,info.PayLoadLen);
				mediainfo->getque.push_back(info);
//				printf("data len = %d\n",m_iDataSize);
				printf("insert getque success, size = %d\n",mediainfo->getque.size());
			}
		}
		mediainfo->rtpmutex_put.Unlock();

		fclose(fp);
//		free(p);
		return 0;
}


int UDPProcessThread::PlayRing()
{
	m_Open_mutex.Lock();
	m_isOpenRing = true;
	m_Open_mutex.Unlock();
	return 0;
}
int UDPProcessThread::StopPlayRing()
{
	m_Open_mutex.Lock();
	m_isOpenRing = false;
	m_Open_mutex.Unlock();
	return 0;
}


int UDPProcessThread::RecvData()
{
	char data[sizeof(RTP_INFO)];
	int ret = 0;
	RTP_INFO info;
	RTPHeader *head = NULL;
	int count = sizeof(RTPHeader);
	int len = sizeof(RTP_INFO);
	
	memset(data, 0, sizeof(data));
	memset(&info,0,sizeof(RTP_INFO));
	int addr_len = sizeof(SOCKADDR);
// 	m_Open_mutex.Lock();
// 	if (m_isOpenRing)
// 	{
// 		OpenRing();
// 	}
// 	m_Open_mutex.Unlock();
//	OpenRing();
	fd_set fdset;
	struct timeval tv;
	FD_ZERO(&fdset);
	FD_SET(m_socket, &fdset);
	tv.tv_sec = 1;
	tv.tv_usec = 0;
	select(FD_SETSIZE, &fdset, 0, 0, &tv);
	if (!FD_ISSET(m_socket, &fdset)){
		if (!m_isSendEndPttInfo){
			m_isSendEndPttInfo = true;
			m_isSendPttInfo = false;
			char t_tcpbuf[32];
			memset(t_tcpbuf, 0, sizeof t_tcpbuf);
			strcpy(t_tcpbuf, "{\"cmd\":4,\"recvPtt\":0}");
			mUDPApi->sendPttTcpInfo(t_tcpbuf);
			mUDPApi->setPttUid(0);
		}
		return -2;
	}
	ret = recvfrom(m_socket,data,sizeof(data),0,(struct sockaddr *)&m_localaddr, &addr_len);
	
	if (ret <= 0)
	{
		printf("recv failed!\n");
		printf("m_socket = %d\n",m_socket);
		Sleep(10);
		return -1;
	}


	head = (RTPHeader *)data;
	//printf("ret = %d payloadtype=%d RTPHeader=%d\n", ret, head->payloadtype, sizeof(RTPHeader));
	if (head->payloadtype != 3 && head->payloadtype != 8 && head->payloadtype != 15){
		return -1;
	}
	//printf("recv from %s:%d\n", inet_ntoa(m_destaddr.sin_addr), ntohs(m_destaddr.sin_port));
	unsigned int t_uid = 0;
	memcpy(&t_uid, data + 11, 4);
	if (!m_isSendPttInfo || m_cur_uid != t_uid){
		m_isSendEndPttInfo = false;
		m_isSendPttInfo = true;
		//zyy add 2017-07-13
		m_cur_uid = t_uid;
		mUDPApi->setPttUid(t_uid);
		//end
		char t_tcpbuf[64];
		memset(t_tcpbuf, 0, sizeof t_tcpbuf);
		sprintf(t_tcpbuf, "{\"cmd\":4,\"recvPtt\":1,\"uid\":%d}", ntohl(t_uid));
		mUDPApi->sendPttTcpInfo(t_tcpbuf);
	}

	info.PayLoadType = head->payloadtype;
	info.Timestamp = head->timestamp;
	info.sNum = head->sequencenumber;
	info.PayLoadLen = ret-11-4;
	info.IsFrame = head->marker;
	memcpy(info.RawData,data+11+4,info.PayLoadLen);

	//printf("===========info PayLoadType=%d sNum=%d PayLoadLen=%d\n", head->payloadtype, head->sequencenumber, info.PayLoadLen);
	//printf("===========info PayLoadType=%d sNum=%d PayLoadLen=%d\n", info.PayLoadType, info.sNum, info.PayLoadLen);
	/*time_t t;
    time(&t);
	FILE *fp=fopen("recvrtp.txt","a");
	fprintf(fp, "%s===========info PayLoadType=%d sNum=%d PayLoadLen=%d\n", ctime(&t), info.PayLoadType, info.sNum, info.PayLoadLen);
	fclose(fp);*/
	m_recv_mutex.Lock();
	mediainfo->rtpmutex_put.Lock();
	//if (m_isRecv)
	{
		mediainfo->rtpque_put.push_back(info);
	}
	mediainfo->rtpmutex_put.Unlock();
	m_recv_mutex.Unlock();
	
	return 0;
}

int UDPProcessThread::SendData()
{
	char data[sizeof(RTP_INFO)];
	int ret;
	RTP_INFO info;
	RTPHead head;
	int len = sizeof(RTP_INFO);
	isockaddr = sizeof(SOCKADDR);
	memset(&info,0,sizeof(info));	
	memset(data,0,sizeof(data));
	memset(&head,0,sizeof(head));
	m_send_mutex.Lock();
	mediainfo->rtpmutex_put.Lock();
//	printf("rtpque_get11 = %d\n",mediainfo->rtpque_get.size());
	
	if (mediainfo->rtpque_get.size()<=0)
	{
		mediainfo->rtpmutex_put.Unlock();
		m_send_mutex.Unlock();
		//zyy add 2017-09-12
		if (!m_isSend)
		{
			if (isSendFlag)
			{
				isSendFlag = false;
				for (int sF = 0; sF < 8; sF++)
				{
					head.num = htons(m_num++);
					head.timestamp = htonl(m_timestamp);
					m_timestamp = m_timestamp + 160;
					head.type = 101;
					head.ssrc = m_rtp_heatbeat_info._session_id;
					memcpy(data, &head, sizeof(head));
					data[11] = 2;
					ret = sendto(m_socket, data, 27, 0, (SOCKADDR *)&m_destaddr, sizeof(m_destaddr));
					memset(&head, 0, sizeof(head));
				}
			}
		}
		//end
		Sleep(10);
		return 0;
	}
	memcpy(&info,&(mediainfo->rtpque_get.front()),sizeof(RTP_INFO));
	mediainfo->rtpque_get.pop_front();
	mediainfo->rtpmutex_put.Unlock();
	if (m_isSend)
	{
		//zyy add 2017-09-12
		if (!isSendFlag)
		{
			isSendFlag = true;
			for (int sF = 0; sF < 8; sF++)
			{
				head.num = htons(m_num++);
				head.timestamp = htonl(m_timestamp);
				m_timestamp = m_timestamp + 160;
				head.type = 101;
				head.ssrc = m_rtp_heatbeat_info._session_id;
				memcpy(data, &head, sizeof(head));
				data[11] = 1;
				ret = sendto(m_socket, data, 27, 0, (SOCKADDR *)&m_destaddr, sizeof(m_destaddr));
				memset(&head, 0, sizeof(head));
			}
		}
		//end
		//head.v = 0x80;
		head.num = htons(m_num++);
		head.timestamp = htonl(m_timestamp);
		m_timestamp = m_timestamp + 160;
		head.type = info.PayLoadType;
		head.ssrc = m_rtp_heatbeat_info._session_id;
		memcpy(data, &head, sizeof(head));
		memcpy(data+11, &m_rtp_heatbeat_info._user_id, 4);
		memcpy(data + 11 + 4, info.RawData, info.PayLoadLen);
	//	printf("enter send\n");
		ret = sendto(m_socket, data, info.PayLoadLen + 11+4, 0, (SOCKADDR *)&m_destaddr, sizeof(m_destaddr));
		if (ret < 0)
		{
			GetLastError();
			m_send_mutex.Unlock();
			printf("ret = sendto:%d\n", ret);
			Sleep(10);
			return -1;
		}
		//printf("rtp media send to %s:%d\n", inet_ntoa(m_destaddr.sin_addr),ntohs(m_destaddr.sin_port));
		/*time_t t;
		time(&t);
		FILE *fp=fopen("sendrtp.txt","a");
		fprintf(fp, "%s===========rtp media send to %s:%d\n", ctime(&t), inet_ntoa(m_destaddr.sin_addr),ntohs(m_destaddr.sin_port));
		fclose(fp);*/
	}
	else
	{
		//zyy add 2017-09-12
		if (isSendFlag)
		{
			isSendFlag = false;
			for (int sF = 0; sF < 8; sF++)
			{
				head.num = htons(m_num++);
				head.timestamp = htonl(m_timestamp);
				m_timestamp = m_timestamp + 160;
				head.type = 101;
				head.ssrc = m_rtp_heatbeat_info._session_id;
				memcpy(data, &head, sizeof(head));
				data[11] = 1;
				ret = sendto(m_socket, data, 27, 0, (SOCKADDR *)&m_destaddr, sizeof(m_destaddr));
				memset(&head, 0, sizeof(head));
			}
		}
		//end
	}
	m_send_mutex.Unlock();
	

	

//	printf("data size =%d\n ",strlen(info.RawData));
	return 0;
}

int UDPProcessThread::SendRtpHeatbeatData(){
	if (!m_isSendHeatbeat){
		return 0;
	}
	char data[sizeof(RTP_INFO)];
	int ret;
	RTPHead head;
	int len = sizeof(RTP_INFO);
	isockaddr = sizeof(SOCKADDR);
	memset(data, 0, sizeof(data));
	memset(&head, 0, sizeof(head));
	//head.v = 0x80;
	head.num = 0;
	//	head.timestamp = info.timeStamp;
	head.type = 50;
	head.ssrc = m_rtp_heatbeat_info._session_id;
	memcpy(data, &head, sizeof(head));
	memcpy(data + 11, &m_rtp_heatbeat_info, sizeof(m_rtp_heatbeat_info));

	ret = sendto(m_socket, data, sizeof(m_rtp_heatbeat_info)+12, 0, (SOCKADDR *)&m_destaddr, sizeof(m_destaddr));
	if (ret < 0)
	{
		GetLastError();
		return -1;
	}
	//printf("rtp heartbeat send to %s:%d\n", inet_ntoa(m_destaddr.sin_addr), ntohs(m_destaddr.sin_port));
	/*time_t t;
    time(&t);
	FILE *fp=fopen("sendheartbeatrtp.txt","a");
	fprintf(fp, "%s======rtp heartbeat send to %s:%d\n", ctime(&t), inet_ntoa(m_destaddr.sin_addr), ntohs(m_destaddr.sin_port));
	fclose(fp);*/

	return 0;
}

int UDPProcessThread::SetRtpHeatbeatInfo(unsigned int sid, unsigned int cid, unsigned int uid){
	m_rtp_heatbeat_info._session_id = htonl(sid);
	m_rtp_heatbeat_info._call_id = htonl(cid);
	m_rtp_heatbeat_info._user_id = htonl(uid);
	m_isSendHeatbeat = true;
	SendRtpHeatbeatData();
	return 0;
}


int UDPProcessThread::ResetRtpHeatbeatInfo(){
	m_rtp_heatbeat_info._session_id = 0;
	m_rtp_heatbeat_info._call_id = 0;
	m_rtp_heatbeat_info._user_id = 0;
	m_isSendHeatbeat = false;
	return 0;
}

int UDPProcessThread::SetRecv(bool flag)
{
	m_recv_mutex.Lock();
	m_isRecv = flag;
	if (!flag)
	{
		mediainfo->rtpmutex_put.Lock();
		mediainfo->rtpque_put.clear();
		mediainfo->getque.clear();
		mediainfo->rtpmutex_put.Unlock();
	}
	m_recv_mutex.Unlock();
	return 0;
}

int UDPProcessThread::SetSend(bool flag)
{
	m_send_mutex.Lock();
	m_num = 1;
	m_isSend = flag;
	m_send_mutex.Unlock();
	return 0;
}

void UDPProcessThread::Close()
{
	CloseSocket();
}

void UDPProcessThread::CloseSocket()
{
	closesocket(m_socket);
}



